package analysis;


import beans.MarketingUserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author zkq
 * @date 2022/10/5 23:32
 */
public class AppMarketingStatistics {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<MarketingUserBehavior> inputStream = env
                .addSource(new SimulatedMarketingBehaviorSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<MarketingUserBehavior>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<MarketingUserBehavior>() {
                            @Override
                            public long extractTimestamp(MarketingUserBehavior element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        })
                );
        inputStream
                .filter(data -> !"UNINSTALL".equals(data.getBehavior()))
                .map(data -> Tuple2.of("total",1))
                .returns(new TypeHint<Tuple2<String, Integer>>() {
                })
                .keyBy(data -> data.f0)
                .window(SlidingEventTimeWindows.of(Time.hours(1),Time.seconds(5)))
                .sum(1)
                .print();
        env.execute();

    }
}

